package com.atguigu.chapter09;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * Author: Pepsi
 * Date: 2023/8/22
 * Desc:
 */
public class Flink05_CEP_Loop_Continues {
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);


        // 1. 定义一个流
        SingleOutputStreamOperator<WaterSensor> stream = env
                .readTextFile("input/sensor.txt")
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((ws, ts) -> ws.getTs())
                );

        // 2. 定义一个模式
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                .<WaterSensor>begin("s1")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_1".equals(value.getId());
                    }
                })
//                .times(2)  // 连续性：默认 followBy 也就是中间可以有间隔
//                .consecutive();  // 加上这个就连续了，必须是两个sensor_1紧紧挨在一起才行
                .times(2)
                .allowCombinations();  // 这两个组合在一起的意思是sensor_1重复两次，不用紧挨，第一个会和后面的每个都结合一次

        // 3. 用CEP类，将模式应用到流中，得到模式流
        PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);

        // 4. 从模式流中选择数据
        ps.select(new PatternSelectFunction<WaterSensor, String>() {

            // 参数是一个map集合，key是模式名字 value是匹配到的所有数据：List<WaterSensor>，为啥是个list? 因为匹配成功一次有可能是多个数据，比如两个数据
            // 这个选择方法是：每匹配成功一次就执行一次，将该次匹配成功的结果pattern，进行自己的逻辑处理后放入到流中
            // flink会把返回值放入到一个新的流中
            @Override
            public String select(Map<String, List<WaterSensor>> pattern) throws Exception {
                return pattern.toString();
            }
        })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
